package it.learn.kafka.producer81.service;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.ObjectFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaProducerException;
import org.springframework.kafka.core.KafkaSendCallback;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

import javax.annotation.PostConstruct;
import java.util.concurrent.ExecutionException;

// 异步发送监听器
@Service
public class Test02Service {

    @Autowired
    private ObjectFactory<KafkaTemplate<String, String>> template;

    // 异步发送
    // @PostConstruct
    public void test01() {
        String data = "测试异步发送代码...";

        ProducerRecord<String, String> record = new ProducerRecord<>("topic1", data);

        ListenableFuture<SendResult<String, String>> future = template.getObject().send(record);
        future.addCallback(new KafkaSendCallback<String, String>() {

            @Override
            public void onSuccess(SendResult<String, String> result) {
                System.out.println("sendResult = " + result.toString());
            }

            @Override
            public void onFailure(KafkaProducerException ex) {
                ProducerRecord<Object, Object> failedProducerRecord = ex.getFailedProducerRecord();
                System.out.println("failedProducerRecord = " + failedProducerRecord.toString());
            }

        });

    }

    // 同步发送
    // @PostConstruct
    public void test02() {
        String data = "测试同步发送代码...";

        ProducerRecord<String, String> record = new ProducerRecord<>("topic1", data);

        try {
            SendResult<String, String> sendResult = template.getObject().send(record).get();
            System.out.println("sendResult = " + sendResult.toString());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

    }

}
